﻿using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using Microsoft.WindowsAzure.ServiceRuntime;
using System.Diagnostics.Contracts;
using System.Reactive;
using System.Reactive.Disposables;
using System.ServiceModel;
using System.Reactive.Linq;

namespace RxCloud
{
    /// <summary>
    /// Represents queue notificationsClient server
    /// </summary>
    public sealed class NotificationsServer
    {
        private readonly RoleInstance roleInstance;
        private readonly string endpointName;
        private readonly string servicePath;

        RefCountDisposable hostDisposable;
        QueueNotifier serviceInstance;


        /// <summary>
        /// Initializes queue notificationsClient server class
        /// </summary>
        public NotificationsServer()
            : this(Constants.DefaultEndpointName)
        {

        }


        /// <summary>
        /// Initializes queue notificationsClient server class
        /// </summary>
        /// <param name="endpointName">Internal communication endpoint name</param>
        public NotificationsServer(string endpointName)
            : this(RoleEnvironment.CurrentRoleInstance, endpointName)
        {
            
        }
        
        /// <summary>
        /// Initializes queue notificationsClient server class
        /// </summary>
        /// <param name="instance">Current role instance</param>
        /// <param name="endpointName">Internal communication endpoint name</param>
        public NotificationsServer(RoleInstance instance, string endpointName)
            : this(instance, endpointName, Constants.DefaultEndpointName)
        { }


        /// <summary>
        /// Initializes queue notificationsClient server class
        /// </summary>
        /// <param name="instance">Current role instance</param>
        /// <param name="endpointName">Internal communication endpoint name</param>
        /// <param name="servicePath">Relative service uri</param>
        public NotificationsServer(RoleInstance instance, string endpointName, string servicePath)
        {
            Contract.Requires(instance != null);
            Contract.Requires(endpointName != null);
            Contract.Requires(servicePath != null);
            Contract.Ensures(this.roleInstance != null);
            Contract.Ensures(this.endpointName != null);
            Contract.Ensures(this.servicePath != null);


            this.roleInstance = instance;
            this.endpointName = endpointName;
            this.servicePath = servicePath;
        }

        /// <summary>
        /// Get notificationsClient stream
        /// </summary>
        /// <param name="queueName">Queue name</param>
        /// <returns>Notifications stream</returns>
        public IObservable<Unit> GetQueueNotification(string queueName)
        {
            Contract.Requires(queueName != null);
            Contract.Ensures(Contract.Result<IObservable<Unit>>() != null);

            //Fault early when endpoint name is invalid
            var endpoint = this.roleInstance.InstanceEndpoints[this.endpointName];
            if (endpoint == null)
            {
                throw new InvalidOperationException(string.Format("No endpoint with name \"{0}\"", this.endpointName));
            }

            return Observable.Create<Unit>(obs =>
            {
                EnsureHost(endpoint);
                var notifications = GetQueueNotifications(this.serviceInstance, queueName);
                var childDisposable = this.hostDisposable.GetDisposable();

                //hostDisposable will not be disposed until all childDisposables not disposed
                this.hostDisposable.Dispose();

                return new CompositeDisposable(notifications.Subscribe(obs), childDisposable);
            });
            
        }

        private void EnsureHost(RoleInstanceEndpoint roleInstanceEndpoint)
        {
            Contract.Requires(roleInstanceEndpoint != null);
            Contract.Ensures(this.hostDisposable != null);
            Contract.Ensures(this.serviceInstance != null);
            Contract.Ensures(!this.hostDisposable.IsDisposed);

            if (this.serviceInstance == null)
            {
                this.serviceInstance = new QueueNotifier();
            }

            if (this.hostDisposable == null 
                || this.hostDisposable.IsDisposed)
            {
                var host = CreateServiceHost(this.serviceInstance, roleInstanceEndpoint, this.servicePath);
                this.hostDisposable = new RefCountDisposable(host);
                Contract.Assume(!this.hostDisposable.IsDisposed);
            }
        }

        private static IObservable<Unit> GetQueueNotifications(QueueNotifier service, string queueName)
        {
            Contract.Requires(service != null);
            Contract.Requires(queueName != null);
            Contract.Ensures(Contract.Result<IObservable<Unit>>() != null);

            return Observable.Create<Unit>(obs =>
            {
                var sub1 = service.MoreMessages
                                  .Where(q => q == queueName)
                                  .Subscribe(q => obs.OnNext(Unit.Default));

                var sub2 = service.QueueCompleted
                                  .Where(q => q == queueName)
                                  .Subscribe(q => obs.OnCompleted());

                return new CompositeDisposable(sub1, sub2);
            });
        }

        private static ServiceHost CreateServiceHost(QueueNotifier queueNotifier, RoleInstanceEndpoint endpoint, string servicePath)
        {
            Contract.Requires(queueNotifier != null);
            Contract.Requires(endpoint != null);
            Contract.Ensures(Contract.Result<ServiceHost>() != null);
            Contract.Ensures(Contract.Result<ServiceHost>().State == CommunicationState.Opened);

            var result = new ServiceHost(queueNotifier);
            var endpointAddress = string.Format(Constants.ServiceEndpointAddress, endpoint.IPEndpoint, servicePath, Utils.GetScheme(endpoint.Protocol));
            result.AddServiceEndpoint(typeof(IQueueNotifier), Utils.GetBinding(endpoint.Protocol), endpointAddress);
            result.Open();
            Contract.Assume(result.State == CommunicationState.Opened);
            return result;
        }
    }
}
